#include <fcntl.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>

#include <iostream>
#include <string>
#include <algorithm>

#include <v8.h>
#include <v8-debug.h>

#include <hadoop/Pipes.hh>
#include <hadoop/TemplateFactory.hh>
#include <hadoop/StringUtils.hh>

#include "jsmr.h"
extern "C" {
  #include "ext/urlparser/urlparser.h"
}


// TODO(crutcher): kill this. HADOOP_BIN_PATH should come from the config
// object or a flag, not a compile time variable.
#define QUOTED_(x) #x
#define QUOTED(x) QUOTED_(x)

/**
 * This executes a command on the local system, capturing the entirity of its
 * standard out to a string which is then returned.
 * @param cmd path and arguments to a command to run
 * @return output of running the command
 */
std::string Exec(std::string cmd) {
  FILE *pipe = popen(cmd.c_str(), "r");
  const unsigned int SIZE = 256;
  char buffer[SIZE] = {0};
  std::string result = "";

  if(pipe == NULL)
    return ""; // Should raise an error

  // TODO(nathan): Make sure that this will handle '\0' correctly.
  while(!feof(pipe)) {
    if(fgets(buffer, SIZE, pipe) != NULL)
      result += buffer;  // is there a more efficient way to do this?
  }

  pclose(pipe);

  return result;
}

/**
 * This is a sub-class of the abstract class ProtocolHandler, specifically for
 * reading files from the Hadoop File System.  It is associated with the 
 * protocol name 'hdfs'
 */
class HDFSHandler : public JSMR::ProtocolHandler {
 public:
  /**
   * This is a sub-class of the abstract class ProtocolHandler, specifically for
   * reading files from the Hadoop File System.  
   * It is associated with the protocol name 'hdfs'
   */
  HDFSHandler() : ProtocolHandler("hdfs") { }

  /**
   * Read a file from the Hadoop File System.  Does so by calling 
   *    hadoop fs -cat $FILE
   * @param long_path The full path to the file from HDFS
   * @return contents of the file read.
   */
  virtual v8::Handle<v8::String> ReadFile(const std::string &long_path)  {
    ::URL storage;
    ::parseURLField(long_path.c_str(), &storage, URL_PATH);

    std::string path(readURLField(long_path.c_str(), storage.path));
    std::string uri = std::string(QUOTED(HADOOP_BIN_PATH)) + " fs -cat " + path;
    std::string data = Exec(uri);

    // TODO(nathan) This is wrong.  Fix this
    if (data == "") {
      // TODO(nathan) Push JSMRLOG into here
      std::cerr << QUOTED(HADOOP_BIN_PATH) << std::endl;
      std::cerr << "Script file not found: " << path << std:: endl;
      abort();
    }

    // TODO(nathan) Is there a cleaner way to cast this?
    char *data_volitile = (char*) malloc(sizeof(char) * data.length());

    bcopy(data.c_str(), data_volitile, data.length());

    v8::Handle<v8::String> result = v8::String::NewExternal(
      new JSMR::CharBufferStringResource(data_volitile, data.length()));

    return result;
  }
};

/**
 * HDFSOperationEnvironment is a subclass of OperationEnvironment which fills in
 * some required methods such as 'Register' and 'ReadFile'.
 *
 * This is an empty constructor.
 */ 
class HDFSOperationEnvironment : public JSMR::OperationEnvironment {
 public:
  /**
   * HDFSOperationEnvironment is a subclass of OperationEnvironment which fills in
   * some required methods such as 'Register' and 'ReadFile'.
   */ 
  HDFSOperationEnvironment() {}

  /**
   * This loads protocol handlers into the environment.  For example, calling with
   * an instance of 'HDFSHandler' will allow 'require' to read from HDFS using the
   * protocol name 'hdfs'.  
   * @param handler A pointer to a protocol handler to register
   */
  virtual void Register(JSMR::ProtocolHandler *handler) {
    if(handlers_.find(handler->GetProtocolName()) != handlers_.end()) {
      Log(JSMR::FATAL, __FILE__, __LINE__,
          "Handler already registered: " + handler->GetProtocolName());
    }
    handlers_[handler->GetProtocolName()] = handler;
  }

  /**
   * Reads a file from the appropriate protocol.  If the file is a relative path
   * then it is searched for along the search path.
   * @param search_paths A vector of strings that define the places so search for
   * the relative path.
   * @param path the path to the file. May be either full path or relative path.
   * @return string contents of the file
   */
  virtual v8::Handle<v8::String> ReadFile(
      const std::vector<std::string> &search_paths,
      const std::string &path) {
    ::URL storage;
    parseURLField(path.c_str(), &storage, URL_SCHEME);
    std::string protocol_name = readURLField(path.c_str(), storage.scheme);
    //path.substr(0, colon_position);

    if(protocol_name == "") {
      //TODO(nathan, crutcher): THIS IS WHERE TO ADD RELATIVE PATHS.
    }

    std::transform(protocol_name.begin(), protocol_name.end(),
        protocol_name.begin(), ::tolower);

    if(handlers_.find(protocol_name) == handlers_.end()) {
      Log(JSMR::FATAL, __FILE__, __LINE__,
          "Unknown protocol: " + protocol_name);
    }

    return handlers_[protocol_name]->ReadFile(path);
  }

 private:
  /**
   * A map from protocol name to protocol handler
   */
  std::map <std::string, JSMR::ProtocolHandler*> handlers_; 
};


class MapContextMapContext : public JSMR::MapContext {
 public:
  MapContextMapContext(HadoopPipes::MapContext *context): context_(context) { }

  virtual const std::string &GetInput() { 
    return context_->getInputValue();
  }

  virtual void Emit(const std::string &key, const std::string &value) {
    context_->emit(key, value);
  } 

 private:
  DISALLOW_COPY_AND_ASSIGN(MapContextMapContext);

  HadoopPipes::MapContext *context_; 
};

class ReduceContextReduceContext : public JSMR::ReduceContext {
 public:
  ReduceContextReduceContext(HadoopPipes::ReduceContext *context)
      : context_(context) { }

  virtual const std::string &GetKey() { 
    return context_->getInputKey();
  }

  virtual const std::string &GetInput() { 
    return context_->getInputValue();
  }

  virtual bool NextValue() {
    return context_->nextValue();
  }

  virtual void Emit(const std::string &key, const std::string &value) {
    context_->emit(key, value);
  } 

 private:
  DISALLOW_COPY_AND_ASSIGN(ReduceContextReduceContext);

  HadoopPipes::ReduceContext *context_; 
};


class JSMRMap: public HadoopPipes::Mapper {
 public:
  JSMRMap(JSMR::OperationEnvironment *env,
          HadoopPipes::TaskContext &context) : env_(env) {
    const std::string *config;
    if (context.getJobConf()->hasKey("jsmr.map.config")) {
      config = &context.getJobConf()->get("jsmr.map.config");
    } else if (context.getJobConf()->hasKey("jsmr.config")) {
      config = &context.getJobConf()->get("jsmr.config");
    } else {
      env_->Log(
        JSMR::FATAL,
        __FILE__, __LINE__,
        "No jsmr.map.config or jsmr.config found in job config");
    }
    container_ = new JSMR::JSContainer(env_, *config);
    container_->Boot();
  }

  ~JSMRMap() {
    delete container_;
  }
  
  void map(HadoopPipes::MapContext& context) {
    MapContextMapContext context_wrapper(&context);
    container_->RunMap(&context_wrapper);
  }

 private:
  DISALLOW_COPY_AND_ASSIGN(JSMRMap);

  JSMR::OperationEnvironment *env_;
  JSMR::JSContainer *container_;
};

class JSMRReduce: public HadoopPipes::Reducer {
 public:
  JSMRReduce(JSMR::OperationEnvironment *env,
             HadoopPipes::TaskContext &context) : env_(env) {
    const std::string *config;
    if (context.getJobConf()->hasKey("jsmr.reduce.config")) {
      config = &context.getJobConf()->get("jsmr.reduce.config");
    } else if (context.getJobConf()->hasKey("jsmr.config")) {
      config = &context.getJobConf()->get("jsmr.config");
    } else {
      env_->Log(
        JSMR::FATAL,
        __FILE__, __LINE__,
        "No jsmr.reduce.config or jsmr.config found in job config");
    }
    container_ = new JSMR::JSContainer(env_, *config);
    container_->Boot();
  }

  ~JSMRReduce() {
    delete container_;
  }
  
  void reduce(HadoopPipes::ReduceContext& context) {
    ReduceContextReduceContext context_wrapper(&context);
    container_->RunReduce(&context_wrapper);
  }

 private:
  DISALLOW_COPY_AND_ASSIGN(JSMRReduce);

  JSMR::OperationEnvironment *env_;
  JSMR::JSContainer *container_;
};

class JSMRFactory : public HadoopPipes::Factory {
 public:
  JSMRFactory(JSMR::OperationEnvironment *env) : env_(env) {}
  ~JSMRFactory() {}

  virtual HadoopPipes::Mapper* createMapper(HadoopPipes::MapContext& context) const {
    return new JSMRMap(env_, context);
  }

  virtual HadoopPipes::Reducer* createReducer(HadoopPipes::ReduceContext& context) const {
    return new JSMRReduce(env_, context);
  }

 private:
  JSMR::OperationEnvironment *env_;

  DISALLOW_COPY_AND_ASSIGN(JSMRFactory);
};

int main(int argc, char* argv[]) {
  HDFSOperationEnvironment env;
  env.Register(new HDFSHandler());
  env.Register(new JSMR::LocalFileHandler());
  JSMRFactory factory(&env);
  int result = HadoopPipes::runTask(factory);
  v8::V8::Dispose();
  return result;
}
